package com.study.flink.word

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * nc -lk 9999
  * 流处理（滑动窗口计算）
  * 每隔一秒统计最近两秒的数据
  *
  * @author: stephen.shen
  * @create: 2019-01-25 14:07
  */
object ScalaStreamWordCount {

  def main(args: Array[String]): Unit = {

    val delimiter = '\n'

    val tool = ParameterTool.fromArgs(args)

    val port: Int = try {
      tool.getInt("port")
    } catch {
      case e: Exception => {
        System.err.println("No port specified. Use default 9999. If you need set, please run 'ScalaWordCount --port <port>'")
      }
        9999
    }

    val hostname: String = try {
      tool.get("hostname")
    } catch {
      case e: Exception => {
        System.err.println("No port specified. Use default 127.0.0.1. If you need set, please run 'ScalaWordCount --hostname <hostname>'")
      }
        "127.0.0.1"
    }

    //添加隐式转换,否则会报错
    import org.apache.flink.api.scala._

    // 获取运行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 连接socket获取数据
    val text = env.socketTextStream(hostname, port, delimiter)
    //val text = env.fromElements("a","v","c","a")
    //val text = env.fromElements(Tuple2("a",1),Tuple2("b",2))
    //val text = env.fromElements(List("a","v","c","a"),List("a","v","c","a"))
    //val text = env.fromCollection(Array("a","v","c","a"))
    //val text = env.fromCollection(List("a","v","c","a"))

    // 解析数据(把数据打平），分组,窗口计算，并且聚合求sum
    val windowCount = text.flatMap(_.split("\\s"))
      .map(w => WordWithCount(w, 1))
      .keyBy("word")
      .timeWindow(Time.seconds(2), Time.seconds(1))
      .sum("count")

    // 设置并行度为1
    windowCount.print().setParallelism(1)
    // 注意：因为flink是懒加载的，所以必须调用execute方法，上面的代码才会执行
    env.execute("Socket window count by scala")

  }

  // case 定义的类可以直接调用，不用new
  case class WordWithCount(word: String, count: Long)

}
